package main.java.LeetCode.ClassicalAlgorithm.Worker.Impl.chain;

import main.java.LeetCode.ClassicalAlgorithm.Queue.Impl.LinkedBlockQueue;
import main.java.LeetCode.ClassicalAlgorithm.Queue.Queue;
import main.java.LeetCode.ClassicalAlgorithm.Worker.Chain;
import main.java.LeetCode.ClassicalAlgorithm.Worker.Filter;

public class GlobalChain<K,V> implements Runnable, Chain<K,V> {

    private Queue<K> in;
    private final Queue<V> out;
    private final Filter<K,V> filter;
    private final Thread thread;
    private final ThreadNode threadNode;

    public GlobalChain(Queue<K> in,Queue<V> out,Filter<K,V> filter){
        this.in = in;
        this.out = out;
        this.filter = filter;
        this.thread = new Thread(this);
        threadNode  = new ThreadNode(this.thread);
    }

    public GlobalChain(Queue<V> out,Filter<K,V> filter){
        this(null,out,filter);
    }

    public GlobalChain(Filter<K,V> filter){
        this(new LinkedBlockQueue<>(Integer.MAX_VALUE),filter);
    }
    public Queue<V> getOut(){
        return out;
    }

    public void setIn(Queue<K> in){
        this.in = in;
    }
    public void setNextThreadNode(ThreadNode t){
        this.threadNode.setNext(t);
    }
    public ThreadNode getThreadNode() {
        return this.threadNode;
    }

    @Override
    public void flow() {
        ThreadNode node = this.threadNode;

        while (node!=null){
            node.getThread().start();
            node = node.getNext();
        }
    }


    @Override
    public void run() {
        K in;
        int count = 0;
        while (!this.thread.isInterrupted()) {
            try {
                in = this.in.get();
                V v = this.filter.doFilter(in);
                this.out.set(v);
                count++;
                if(count>=this.getThreadNode().getSize())
                    break;
            } catch (Exception e) {
                this.thread.interrupt();
            }
        }
    }

    @Override
    public <E> GlobalChain<V, E> linkGlobalChain(GlobalChain<V, E> chain) {
        chain.setIn(this.out);
        this.setNextThreadNode(chain.getThreadNode());
        chain.getThreadNode().setSize(this.threadNode.getSize());
        return chain;
    }
}
